package com.tencreat.mq;

import com.rabbitmq.client.Channel;
import com.tencreat.common.enums.BalanceBusinessType;
import com.tencreat.common.enums.BalanceTypeEnum;
import com.tencreat.common.enums.mall.AllEnum;
import com.tencreat.common.utils.bean.UserBalanceChange;
import com.tencreat.common.utils.uuid.IdUtils;
import com.tencreat.domain.account.*;
import com.tencreat.domain.mall.BalanceChange;
import com.tencreat.domain.mall.GoodsRollOrder;
import com.tencreat.domain.mall.StoreWithdrawAudit;
import com.tencreat.domain.mall.SupplierWithdrawAudit;
import com.tencreat.service.account.IAccountWithdrawInfoService;
import com.tencreat.service.account.ICommonWithdrawConfigService;
import com.tencreat.service.account.IMallStoreService;
import com.tencreat.service.account.IMallUserService;
import com.tencreat.service.mall.IBalanceChangeService;
import com.tencreat.service.mall.impl.RechargeLimitChangeServiceImpl;
import jodd.util.StringUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

/**
 * 余额、货款、代采购额度、券变动处理
 */
@Service
@Slf4j
//@RequiredArgsConstructor
public class BalanceChangeConsumer {

    @Autowired
    IMallUserService mallUserService;
    @Autowired
    IBalanceChangeService balanceChangeService;
    @Autowired
    RedissonClient redissonClient;
    @Autowired
    PlatformTransactionManager platformTransactionManager;
    @Autowired
    TransactionDefinition transactionDefinition;
    @Autowired
    ICommonWithdrawConfigService withdrawConfigService;
    @Autowired
    IAccountWithdrawInfoService accountWithdrawInfoService;
    @Autowired
    IMallStoreService mallStoreService;
    @Autowired
    RechargeLimitChangeServiceImpl rechargeLimitChangeService;


    @RabbitListener(queues = "base-mall.user.balance.name.v2", concurrency = "1")
    public void doJob(Message message, Channel channel) throws Exception {
        log.info("进入余额变化队列----------------");
        ByteArrayInputStream byteInt = new ByteArrayInputStream(message.getBody());
        ObjectInputStream objInt = new ObjectInputStream(byteInt);
        UserBalanceChange userBalanceMessage = (UserBalanceChange) objInt.readObject();
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        //信息为空,丢弃该信息
        if (Objects.isNull(userBalanceMessage)) {
            log.info("消息为空,手动丢弃该消息....");
//            channel.basicNack(deliveryTag, false, false);
            channel.basicAck(deliveryTag, false);
            return;
        }
        boolean isAdd = userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType());
        log.info("余额变动： 用户：【{}】{}【{}】", userBalanceMessage.getUuid(), isAdd ? "增加" : "扣除", userBalanceMessage.getChangAmount());
        MallUser mallUser = mallUserService.lambdaQuery().eq(MallUser::getUuid, userBalanceMessage.getUuid()).one();
        if (mallUser == null || mallUser.getUuid() == null) {
            log.info("用户【{}】未找到为空,手动丢弃该消息....", userBalanceMessage.getUuid());
//            channel.basicNack(deliveryTag, false, false);
            channel.basicAck(deliveryTag, false);
            return;
        }
        int id = IdUtils.generateUUID();
        // 同一个用户操作加锁
        String lockKey = "user_change_queue_" + userBalanceMessage.getUuid();
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(10, TimeUnit.SECONDS);
        log.info("加锁成功id:{}, key:{}", userBalanceMessage.getUuid(), lockKey);
        // 先锁再开启事务
//        TransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
        TransactionStatus transaction = platformTransactionManager.getTransaction(transactionDefinition);

        // 判断是否重复消费
        if (userBalanceMessage.getSource() != null) {
            List<BalanceChange> list = balanceChangeService.lambdaQuery().eq(BalanceChange::getUuid, mallUser.getUuid())
                    .eq(BalanceChange::getSource, userBalanceMessage.getSource())
                    .eq(BalanceChange::getBusinessType, userBalanceMessage.getBusinessType())
                    .eq(BalanceChange::getBalanceType, userBalanceMessage.getBalanceType())
                    .eq(BalanceChange::getType, userBalanceMessage.getType())
                    .list();
            if (list != null && list.size() > 0) {
                log.info("余额重复消费，手动丢弃该消息-----------------");
                platformTransactionManager.rollback(transaction);
//                    channel.basicNack(deliveryTag, false, false);
                channel.basicAck(deliveryTag, false);
                return;
            }
        }
        try {
            BalanceChange balanceChange = new BalanceChange();
            balanceChange(mallUser, userBalanceMessage, balanceChange);

            platformTransactionManager.commit(transaction);
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            e.printStackTrace();
            platformTransactionManager.rollback(transaction);
            log.info("mq报错id:{}-----------{}", userBalanceMessage.getUuid(), e.getMessage());
        } finally {
            log.info("余额变化mq释放锁id:{}--------------------lockkey:{}", userBalanceMessage.getUuid(), lockKey);
            if (lock.isHeldByCurrentThread()) { // 时候是当前执行线程的锁
                lock.unlock(); // 释放锁
            }
        }
    }
    public void balanceChange(MallUser mallUser, UserBalanceChange userBalanceMessage, BalanceChange balanceChange) {
        balanceChange.setUuid(mallUser.getUuid());
        balanceChange.setType(userBalanceMessage.getType());
        balanceChange.setBusinessType(userBalanceMessage.getBusinessType());
        balanceChange.setCreateTime(LocalDateTime.now());
        balanceChange.setUpdateTime(LocalDateTime.now());
        balanceChange.setRemarks(userBalanceMessage.getRemark());
        balanceChange.setSource(userBalanceMessage.getSource());
        balanceChange.setBalanceType(userBalanceMessage.getBalanceType());

        BigDecimal amount = userBalanceMessage.getChangAmount().abs();
        //余额变动
        if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.BALANCE) {
            balanceChange.setBalanceBefore(mallUser.getBalance());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setBalance(mallUser.getBalance().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setBalance(mallUser.getBalance().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getBalance());
            balanceChangeService.save(balanceChange);
            log.info("余额修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getBalance, mallUser.getBalance()).set(MallUser::getUpdateTime, new Date()).eq(MallUser::getId, mallUser.getId()).update();
        }
        //消费券变动
        else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.CONSUMER_ROLL) {
            balanceChange.setBalanceBefore(mallUser.getConsumerRoll());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setConsumerRoll(mallUser.getConsumerRoll().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setConsumerRoll(mallUser.getConsumerRoll().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getConsumerRoll());
            balanceChangeService.save(balanceChange);
            log.info("消费券修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getConsumerRoll, mallUser.getConsumerRoll()).eq(MallUser::getId, mallUser.getId()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.WAIT_GOODS_ROLL) {
            balanceChange.setBalanceBefore(mallUser.getGoodsRollWait());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setGoodsRollWait(mallUser.getGoodsRollWait().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setGoodsRollWait(mallUser.getGoodsRollWait().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getGoodsRollWait());
            balanceChangeService.save(balanceChange);

            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                //将提货单记录存入goods_roll_order
                GoodsRollOrder goodsRollOrder = new GoodsRollOrder();
                goodsRollOrder.setUuid(mallUser.getUuid());
                goodsRollOrder.setBalanceType(BalanceTypeEnum.WAIT_GOODS_ROLL);
                goodsRollOrder.setWaitGoodsRoll(balanceChange.getChangAmount());
                goodsRollOrder.setBusinessType(userBalanceMessage.getBusinessType());
                goodsRollOrder.setStatus(0);
                goodsRollOrder.setSource(balanceChange.getSource());
                goodsRollOrder.setMobile(mallUser.getMobile());
                goodsRollOrder.setCreateTime(LocalDateTime.now());
                //获取银行卡绑定信息
                AccountWithdrawInfo one = accountWithdrawInfoService.lambdaQuery().eq(AccountWithdrawInfo::getUserId, mallUser.getUuid()).one();
                if (one != null) {
                    goodsRollOrder.setBankCode(one.getBankCode());
                    goodsRollOrder.setBankName(one.getBankName());
                }

                goodsRollOrder.insert();
            }

            log.info("待结算提货单修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getGoodsRollWait, mallUser.getGoodsRollWait()).eq(MallUser::getId, mallUser.getId()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.GOODS_ROLL) {
            balanceChange.setBalanceBefore(mallUser.getGoodsRoll());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setGoodsRoll(mallUser.getGoodsRoll().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setGoodsRoll(mallUser.getGoodsRoll().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getGoodsRoll());
            balanceChangeService.save(balanceChange);

            log.info("提货单修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getGoodsRoll, mallUser.getGoodsRoll()).eq(MallUser::getId, mallUser.getId()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.FINANCE_WAIT) {
            balanceChange.setBalanceBefore(mallUser.getFinanceWait());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setFinanceWait(mallUser.getFinanceWait().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setFinanceWait(mallUser.getFinanceWait().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getFinanceWait());
            balanceChangeService.save(balanceChange);
            log.info("供应商待结算货款修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getFinanceWait, mallUser.getFinanceWait()).eq(MallUser::getId, mallUser.getId()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.FINANCE) {
            balanceChange.setBalanceBefore(mallUser.getFinance());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setFinance(mallUser.getFinance().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setFinance(mallUser.getFinance().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getFinance());
            balanceChangeService.save(balanceChange);
            log.info("供应商货款修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getFinance, mallUser.getFinance()).eq(MallUser::getId, mallUser.getId()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.STORE_FINANCE_WAIT) {
//            MallStore mallStore = mallStoreService.lambdaQuery().eq(MallStore::getUuid, mallUser.getUuid()).one();
            balanceChange.setBalanceBefore(mallUser.getStoreFinanceWait());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setStoreFinanceWait(mallUser.getStoreFinanceWait().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setStoreFinanceWait(mallUser.getStoreFinanceWait().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getStoreFinanceWait());
            balanceChangeService.save(balanceChange);
            log.info("店铺待结算货款修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getStoreFinanceWait, mallUser.getStoreFinanceWait()).eq(MallUser::getId, mallUser.getId()).update();
//            mallStoreService.lambdaUpdate().set(MallStore::getStoreFinanceWait, mallStore.getStoreFinanceWait()).eq(MallStore::getUuid, mallUser.getUuid()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.STORE_FINANCE) {
            balanceChange.setBalanceBefore(mallUser.getStoreFinance());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setStoreFinance(mallUser.getStoreFinance().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setStoreFinance(mallUser.getStoreFinance().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getStoreFinance());
            balanceChangeService.save(balanceChange);
            log.info("店铺货款修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getStoreFinance, mallUser.getStoreFinance()).eq(MallUser::getId, mallUser.getId()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.WAIT_INCOME) {
            balanceChange.setBalanceBefore(mallUser.getWaitIncome());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setWaitIncome(mallUser.getWaitIncome().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setWaitIncome(mallUser.getWaitIncome().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getWaitIncome());
            balanceChangeService.save(balanceChange);
            log.info("待结算收益修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getWaitIncome, mallUser.getWaitIncome()).eq(MallUser::getId, mallUser.getId()).update();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.INCOME) {
            balanceChange.setBalanceBefore(mallUser.getAccruedIncome());
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                balanceChange.setChangAmount(amount);
                mallUser.setAccruedIncome(mallUser.getAccruedIncome().add(amount));
            } else {
                balanceChange.setChangAmount(amount.negate());
                mallUser.setAccruedIncome(mallUser.getAccruedIncome().subtract(amount));
            }
            balanceChange.setBalance(mallUser.getAccruedIncome());
            balanceChangeService.save(balanceChange);
            log.info("累计收益修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getAccruedIncome, mallUser.getAccruedIncome()).eq(MallUser::getId, mallUser.getId()).update();
        }

        if (userBalanceMessage.getBusinessType().equals(BalanceBusinessType.WITHDRAW)
                && userBalanceMessage.getType().equals("2")
                && userBalanceMessage.getBalanceType() == BalanceTypeEnum.BALANCE) {
            log.info("进入提现记录业务-------------" + mallUser.getUuid());
            // 如果是提现申请， 添加提现记录
            CommonWithdrawConfig commonWithdrawConfig = withdrawConfigService.lambdaQuery().eq(CommonWithdrawConfig::getType, 1).one();
            BigDecimal interest = userBalanceMessage.getChangAmount().abs().multiply(commonWithdrawConfig.getHandlingFee());
            BigDecimal end = userBalanceMessage.getChangAmount().abs().subtract(interest);

            AccountWithdrawAudit accountWithdrawAudit = new AccountWithdrawAudit();
            accountWithdrawAudit.setBalanceId(balanceChange.getId());
            accountWithdrawAudit.setUuid(mallUser.getUuid());
            accountWithdrawAudit.setResult(0);
            if (StringUtil.isBlank(userBalanceMessage.getRemark())) {
                userBalanceMessage.setRemark("3");// 银行卡
            }
            accountWithdrawAudit.setWithdrawType(Integer.valueOf(userBalanceMessage.getRemark()));
            accountWithdrawAudit.setValue(end);
            accountWithdrawAudit.setServiceMoney(interest);
            accountWithdrawAudit.setCreateTime(LocalDateTime.now());
            accountWithdrawAudit.setUpdateTime(LocalDateTime.now());
            accountWithdrawAudit.insert();
        } else if (userBalanceMessage.getBusinessType().equals(BalanceBusinessType.WITHDRAW)
                && userBalanceMessage.getType().equals("2")
                && userBalanceMessage.getBalanceType() == BalanceTypeEnum.FINANCE) {
            log.info("进入供应商货款提现记录业务-------------" + mallUser.getUuid());
            CommonWithdrawConfig commonWithdrawConfig = withdrawConfigService.lambdaQuery().eq(CommonWithdrawConfig::getType, 2).eq(CommonWithdrawConfig::getUserType,12).one();
            BigDecimal interest = userBalanceMessage.getChangAmount().abs().multiply(commonWithdrawConfig.getHandlingFee());
            //增加货款提现记录
            SupplierWithdrawAudit supplierWithdrawAudit = new SupplierWithdrawAudit();
            supplierWithdrawAudit.setUuid(mallUser.getUuid());
            supplierWithdrawAudit.setName(mallUser.getUsername() != null ? mallUser.getUsername() : mallUser.getNickname());
            supplierWithdrawAudit.setMobile(mallUser.getMobile());
            supplierWithdrawAudit.setResult(0);
            supplierWithdrawAudit.setWithdrawType(Integer.valueOf(userBalanceMessage.getRemark()));
            supplierWithdrawAudit.setValue(userBalanceMessage.getChangAmount());
            supplierWithdrawAudit.setServiceMoney(interest);
            supplierWithdrawAudit.setCreateTime(LocalDateTime.now());
            supplierWithdrawAudit.setUpdateTime(LocalDateTime.now());
            supplierWithdrawAudit.insert();
        } else if (userBalanceMessage.getBusinessType().equals(BalanceBusinessType.WITHDRAW)
                && userBalanceMessage.getType().equals("2")
                && userBalanceMessage.getBalanceType() == BalanceTypeEnum.STORE_FINANCE) {
            log.info("进入店铺货款提现记录业务-------------" + mallUser.getUuid());
            CommonWithdrawConfig commonWithdrawConfig = withdrawConfigService.lambdaQuery().eq(CommonWithdrawConfig::getType, 2).eq(CommonWithdrawConfig::getUserType,13).one();
            BigDecimal interest = userBalanceMessage.getChangAmount().abs().multiply(commonWithdrawConfig.getHandlingFee());
            //增加货款提现记录
            StoreWithdrawAudit storeWithdrawAudit = new StoreWithdrawAudit();
            storeWithdrawAudit.setUuid(mallUser.getUuid());
            storeWithdrawAudit.setName(mallUser.getUsername() != null ? mallUser.getUsername() : mallUser.getNickname());
            storeWithdrawAudit.setMobile(mallUser.getMobile());
            storeWithdrawAudit.setResult(0);
            storeWithdrawAudit.setWithdrawType(Integer.valueOf(userBalanceMessage.getRemark()));
            storeWithdrawAudit.setValue(userBalanceMessage.getChangAmount());
            storeWithdrawAudit.setServiceMoney(interest);
            storeWithdrawAudit.setCreateTime(LocalDateTime.now());
            storeWithdrawAudit.setUpdateTime(LocalDateTime.now());
            storeWithdrawAudit.insert();
        } else if (userBalanceMessage.getBalanceType() == BalanceTypeEnum.RECHARGE_LIMIT_CHANGE) {
            RechargeLimitChange rechargeLimitChange = new RechargeLimitChange();
            //获取当前用户代采购额度
            rechargeLimitChange.setBalanceBefore(mallUser.getRechargeLimit());
            //直推、间推、团队用户增加代采购额度
            if (userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType())) {
                rechargeLimitChange.setChangAmount(amount);
                mallUser.setRechargeLimit(mallUser.getRechargeLimit().add(amount));
            } else {
                //充值减少代采购额度
                rechargeLimitChange.setChangAmount(amount.negate());
                mallUser.setRechargeLimit(mallUser.getRechargeLimit().subtract(amount));
            }

            rechargeLimitChange.setBalance(mallUser.getRechargeLimit());

            rechargeLimitChange.setUuid(userBalanceMessage.getUuid());
            rechargeLimitChange.setSource(userBalanceMessage.getSource());
            rechargeLimitChange.setType(userBalanceMessage.getType());
            rechargeLimitChange.setBusinessType(userBalanceMessage.getBusinessType());
            rechargeLimitChange.setCreateTime(LocalDateTime.now());
            rechargeLimitChange.setUpdateTime(LocalDateTime.now());
            rechargeLimitChange.setRemarks(userBalanceMessage.getRemark());
            rechargeLimitChange.setBalanceType(userBalanceMessage.getBalanceType());


            rechargeLimitChangeService.save(rechargeLimitChange);
            log.info("代采购额度修改记录------用户id{}-----备注{}---", mallUser.getUuid(), userBalanceMessage.getRemark());
            mallUserService.lambdaUpdate().set(MallUser::getRechargeLimit, mallUser.getRechargeLimit()).eq(MallUser::getId, mallUser.getId()).update();
        }
    }

    public void doJobAsync(UserBalanceChange userBalanceMessage) throws Exception {
        log.info("进入同步余额变化队列----------------");

        // 同一个用户操作加锁
        String lockKey = "user_change_queue_" + userBalanceMessage.getUuid();
        RLock lock = redissonClient.getLock(lockKey);
        lock.lock(10, TimeUnit.SECONDS);

        // 先锁再开启事务
//        TransactionDefinition transactionDefinition = new DefaultTransactionDefinition();
        TransactionStatus transaction = platformTransactionManager.getTransaction(transactionDefinition);

        try {
            boolean isAdd = userBalanceMessage.getType().equals(AllEnum.BalanceChangeTypeEnums.INCOME.getType());
            log.info("余额变动： 用户：【{}】{}【{}】", userBalanceMessage.getUuid(), isAdd ? "增加" : "扣除", userBalanceMessage.getChangAmount());


            MallUser mallUser = mallUserService.lambdaQuery().eq(MallUser::getUuid, userBalanceMessage.getUuid()).one();
            if (mallUser == null) {
                log.info("用户【{}】未找到为空,手动丢弃该消息....", userBalanceMessage.getUuid());
                return;
            }
            // 判断是否重复消费
            if (userBalanceMessage.getSource() != null) {
                List<BalanceChange> list = balanceChangeService.lambdaQuery().eq(BalanceChange::getUuid, mallUser.getUuid())
                        .eq(BalanceChange::getSource, userBalanceMessage.getSource())
                        .eq(BalanceChange::getBusinessType, userBalanceMessage.getBusinessType())
                        .list();
                if (list != null && list.size() > 0) {
                    log.info("余额重复消费，手动丢弃该消息-----------------");
                    return;
                }
            }
            BalanceChange balanceChange = new BalanceChange();
            balanceChange(mallUser, userBalanceMessage, balanceChange);

            platformTransactionManager.commit(transaction);
        } catch (Exception e) {
            e.printStackTrace();
            platformTransactionManager.rollback(transaction);
            log.info("mq报错-----------" + e.getCause());
        } finally {
            log.info("余额变化mq释放锁--------------------" + lockKey);
            if (lock.isHeldByCurrentThread()) { // 时候是当前执行线程的锁
                lock.unlock(); // 释放锁
            }
        }
    }
}
